package com.chhuang.common;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.chhuang.bean.QueryTermBean;

/**
 * 数据库常用方法父类
 * @author HuangChen
 *
 */
public class DataOperateAbstract {
	
	private static final Log LOG = LogFactory.getLog(DataOperateAbstract.class);
	
	protected static final String INSERT_INTO = "insert into ";
	protected static final String DELETE = "delete from ";
	protected static final String UPDATE = "update ";
	protected static final String SELECT = "select ";
	protected static final String FROM = " from ";
	protected static final String TRUNCATE_TABLE = "truncate table ";
	protected static final String COUNT = "count(1) count";
	protected static final String MAX = "max(?) max";
	protected static final String WHERE = " where ";
	protected static final String AND = " and ";
	protected static final String SET = " set ";
	protected static final String VALUES = "values(";
	protected static final String STAR = "*";
	protected static final String LEFT_BRACKET = "(";
	protected static final String RIGHT_BRACKET = ")";
	protected static final String QUEST = "?";
	protected static final String COMMA = ",";
	protected static final String EQUAL = "=";
	protected static final String MORE = ">";
	protected static final String LESS = "<";
	
	protected static final int AND_LEN = AND.length();
	protected static final int COMMA_LEN = COMMA.length();
	
	protected static final String CHAR = "char";
	protected static final String TEXT = "text";
	protected static final String NUMBER = "number";
	protected static final String DECIMAL = "decimal";
	protected static final String TIME = "time";
	protected static final String DATE = "date";
	protected static final String BLOB = "blob";
	protected static final String CLOB = "clob";
	
	protected static final int EXECUTE_NUM = 1000; 
			
	protected C3p0Helper c3p0Helper = C3p0Helper.getInstance();

	/**
	 * 查找
	 * @param connection
	 * @param table
	 * @param queryBean 查询条件
	 * @param selectColumns 需要查询字段，null表示查询所有字段
	 * @return
	 */
	public PreparedStatement find(Connection connection, String table, QueryTermBean queryBean, String... selectColumns){
		PreparedStatement ps=null;
		if(connection==null || Assert.isInvalidString(table)){
			return ps;
		}

		StringBuilder afterSelect = new StringBuilder();
		if(Assert.isInvalidArray(selectColumns)){
			afterSelect.append(STAR);
		}else{
			for(String column : selectColumns){
				afterSelect.append(column).append(COMMA);
			}
			int aslen = afterSelect.length();
			afterSelect.delete(aslen-COMMA_LEN, aslen);
		}
		
		StringBuilder where = new StringBuilder();
		if(queryBean!=null && Assert.isValidString(queryBean.getTermStr())){
			where.append(WHERE).append(queryBean.getTermStr());
		}
		
		StringBuilder sql = new StringBuilder(SELECT).append(afterSelect).append(FROM).append(table).append(where);
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql.toString());
			if(queryBean!=null && Assert.isValidArray(queryBean.getTermValue())){
				for(int i=0;i<queryBean.getTermValue().length;i++){
					ps.setObject((i+1), queryBean.getTermValue()[i]);
				}
			}
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return ps;
	}
	
	/**
	 * 查找
	 * @param connection
	 * @param sql
	 * @return
	 */
	public PreparedStatement query(Connection connection, String sql){
		PreparedStatement ps=null;
		if(connection==null || Assert.isInvalidString(sql)){
			return ps;
		}
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql);
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return ps;
	}
	
	/**
	 * 查找
	 * @param connection
	 * @param sql
	 * @param args
	 * @return
	 */
	public PreparedStatement query(Connection connection, String sql, Object... args){
		PreparedStatement ps=null;
		if(connection==null || Assert.isInvalidString(sql)){
			return ps;
		}
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql);
			if(Assert.isValidArray(args)){
				for(int i=0;i<args.length;i++){
					ps.setObject((i+1), args[i]);
				}
			}
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return ps;
	}
	
	/**
	 * 保存
	 * @param connection
	 * @param table
	 * @param rs
	 * @param modifyTimeColumn 修改时间字段名称，增量同步时用来检测修改时间是不是大于当前时间，为空时表示不需要检测
	 * @return
	 */
	public boolean save(Connection connection, String table, ResultSet rs, String modifyTimeColumn){
		boolean result = false;
		if(connection==null || Assert.isInvalidString(table) || rs==null){
			return result;
		}
		
		PreparedStatement ps = null;
		try {
			if(rs.next()){
				ResultSetMetaData rsmd = rs.getMetaData();
				int columnCount = rsmd.getColumnCount();
				if(rsmd!=null && columnCount>0){
					StringBuilder afterTable = new StringBuilder(LEFT_BRACKET);
					StringBuilder valueStr = new StringBuilder(VALUES);
					for(int i=1;i<=columnCount;i++){
						afterTable.append(rsmd.getColumnName(i).toLowerCase()).append(COMMA);
						valueStr.append(QUEST).append(COMMA);
					}
					int atlen = afterTable.length();
					afterTable.replace(atlen-COMMA_LEN, atlen, RIGHT_BRACKET);
					int vslen = valueStr.length();
					valueStr.replace(vslen-COMMA_LEN, vslen, RIGHT_BRACKET);
					
					String sql = new StringBuilder(INSERT_INTO).append(table).append(afterTable).append(valueStr).toString();
					LOG.debug(sql);
					int n=0;
					ps = connection.prepareStatement(sql);
					ExecutorService pool = Executors.newFixedThreadPool(5);
					do{
						for(int i=1;i<=columnCount;i++){
							String type = rsmd.getColumnTypeName(i).toLowerCase();
							if(Assert.isValidString(type)){
								if(type.indexOf(CHAR)>-1 || type.indexOf(TEXT)>-1){
									ps.setString(i, rs.getString(i));
								}else if(NUMBER.equals(type)){
									ps.setBigDecimal(i, rs.getBigDecimal(i));
								}else if(type.indexOf(TIME)>-1 || type.indexOf(DATE)>-1){
									Timestamp timestamp = rs.getTimestamp(i);
									Timestamp nowTime = new Timestamp(System.currentTimeMillis());
									if(timestamp!=null && Assert.isValidString(modifyTimeColumn) 
											&& modifyTimeColumn.equalsIgnoreCase(rsmd.getColumnName(i))
											&& timestamp.after(nowTime)){//修正晚于当前时间的修改时间乱入
										ps.setTimestamp(i, nowTime);
									}else{
										ps.setTimestamp(i, timestamp);
									}
								}else if(type.indexOf(BLOB)>-1){
									ps.setBlob(i, rs.getBlob(i));
								}else if(CLOB.equals(type)){
									ps.setClob(i, rs.getClob(i));
								}else{
									ps.setObject(i, rs.getObject(i));
								}
							}
						}
						ps.addBatch();
						n++;
						if(n==EXECUTE_NUM){//先执行一批EXECUTE_NUM条
							pool.execute(new PSExecuter(ps,table,n));
							ps = connection.prepareStatement(sql);
							n=0;
						}
					}while(rs.next());
					if(n>0){//最后可能有不足EXECUTE_NUM条的
						pool.execute(new PSExecuter(ps,table,n));
					}
					ThreadPoolShutdownUtil.shutdownPool(pool, 20, TimeUnit.HOURS);
				}
			}
			result = true;
		} catch (SQLException e) {
			e.printStackTrace();
		} finally{
			c3p0Helper.closes(ps);
		}
		
		return result;
	}
	
	/**
	 * 修改
	 * @param connection
	 * @param table
	 * @param idColumns 主键或唯一键，用来确定要修改哪些记录，为空时表示修改所有记录
	 * @param rs
	 * @param modifyTimeColumn 修改时间字段名称，增量同步时用来检测修改时间是不是大于当前时间，为空时表示不需要检测
	 * @return
	 */
	public boolean update(Connection connection, String table, String[] idColumns, ResultSet rs, String modifyTimeColumn){
		boolean result = false;
		if(connection==null || Assert.isInvalidString(table) || rs==null){
			return result;
		}

		PreparedStatement ps = null;
		try {
			if(rs.next()){
				ResultSetMetaData rsmd = rs.getMetaData();
				int columnCount = rsmd.getColumnCount();
				if(rsmd!=null && columnCount>0){
					StringBuilder setStr = new StringBuilder(SET);
					for(int i=1;i<=columnCount;i++){
						setStr.append(rsmd.getColumnName(i).toLowerCase()).append(EQUAL).append(QUEST).append(COMMA);
					}
					int sslen = setStr.length();
					setStr.delete(sslen-COMMA_LEN, sslen);
					
					StringBuilder where = new StringBuilder();
					if(Assert.isValidArray(idColumns)){
						where.append(WHERE);
						for(String idColumn : idColumns){
							where.append(idColumn).append(EQUAL).append(QUEST).append(AND);
						}
						int wlen = where.length();
						where.delete(wlen-AND_LEN, wlen);
					}
					
					String sql = new StringBuilder(UPDATE).append(table).append(setStr).append(where).toString();
					LOG.debug(sql);
					
					int n=0;
					ps = connection.prepareStatement(sql);
					ExecutorService pool = Executors.newFixedThreadPool(5);
					do{
						int i=1;
						for(;i<=columnCount;i++){
							String type = rsmd.getColumnTypeName(i).toLowerCase();
							if(Assert.isValidString(type)){
								if(type.indexOf(CHAR)>-1 || type.indexOf(TEXT)>-1){
									ps.setString(i, rs.getString(i));
								}else if(NUMBER.equals(type) || DECIMAL.equals(type)){
									ps.setBigDecimal(i, rs.getBigDecimal(i));
								}else if(type.indexOf(TIME)>-1 || type.indexOf(DATE)>-1){
									Timestamp timestamp = rs.getTimestamp(i);
									Timestamp nowTime = new Timestamp(System.currentTimeMillis());
									if(timestamp!=null && Assert.isValidString(modifyTimeColumn) 
											&& modifyTimeColumn.equalsIgnoreCase(rsmd.getColumnName(i))
											&& timestamp.after(nowTime)){//修正晚于当前时间的修改时间乱入
										ps.setTimestamp(i, nowTime);
									}else{
										ps.setTimestamp(i, timestamp);
									}
								}else if(type.indexOf(BLOB)>-1){
									ps.setBlob(i, rs.getBlob(i));
								}else if(CLOB.equals(type)){
									ps.setClob(i, rs.getClob(i));
								}else{
									ps.setObject(i, rs.getObject(i));
								}
							}
						}
						if(Assert.isValidArray(idColumns)){
							for(String idColumn : idColumns){
								ps.setObject(i, rs.getObject(idColumn));
								i++;
							}
						}
						ps.addBatch();
						n++;
						if(n==EXECUTE_NUM){//先执行一批EXECUTE_NUM条
							pool.execute(new PSExecuter(ps,table,n));
							ps = connection.prepareStatement(sql);
							n=0;
						}
					}while(rs.next());
					if(n>0){//最后可能有不足EXECUTE_NUM条的
						pool.execute(new PSExecuter(ps,table,n));
					}
					ThreadPoolShutdownUtil.shutdownPool(pool, 20, TimeUnit.HOURS);
				}
			}
			result = true;
		} catch (SQLException e) {
			e.printStackTrace();
		} finally{
			c3p0Helper.closes(ps);
		}
		
		return result;
	}
	
	/**
	 * 保存或者修改
	 * @param connection
	 * @param table
	 * @param idColumns 主键或唯一键，用来确定要修改哪些记录，并且判断是新增还是修改，不能为空
	 * @param rs
	 * @param modifyTimeColumn 修改时间字段名称，增量同步时用来检测修改时间是不是大于当前时间，为空时表示不需要检测
	 * @return
	 */
	public boolean saveOrUpdate(Connection connection, String table, String[] idColumns, ResultSet rs, String modifyTimeColumn){
		boolean result = false;
		if(connection==null || Assert.isInvalidString(table) || Assert.isInvalidArray(idColumns) || rs==null){
			return result;
		}

		PreparedStatement insertPs = null;
		PreparedStatement updatePs = null;
		try {
			if(rs.next()){
				ResultSetMetaData rsmd = rs.getMetaData();
				int columnCount = rsmd.getColumnCount();
				if(rsmd!=null && columnCount>0){
					StringBuilder afterTable = new StringBuilder(LEFT_BRACKET);
					StringBuilder valueStr = new StringBuilder(VALUES);
					StringBuilder setStr = new StringBuilder(SET);
					for(int i=1;i<=columnCount;i++){
						afterTable.append(rsmd.getColumnName(i).toLowerCase()).append(COMMA);
						valueStr.append(QUEST).append(COMMA);
						setStr.append(rsmd.getColumnName(i).toLowerCase()).append(EQUAL).append(QUEST).append(COMMA);
					}
					int atlen = afterTable.length();
					afterTable.replace(atlen-COMMA_LEN, atlen, RIGHT_BRACKET);
					int vslen = valueStr.length();
					valueStr.replace(vslen-COMMA_LEN, vslen, RIGHT_BRACKET);
					int sslen = setStr.length();
					setStr.delete(sslen-COMMA_LEN, sslen);
					
					StringBuilder where = new StringBuilder();
					StringBuilder afterWhere = new StringBuilder();
					for(String idColumn : idColumns){
						afterWhere.append(idColumn).append(EQUAL).append(QUEST).append(AND);
					}
					int awlen = afterWhere.length();
					afterWhere.delete(awlen-AND_LEN, awlen);
					where.append(WHERE).append(afterWhere);
					
					String insertSql = new StringBuilder(INSERT_INTO).append(table).append(afterTable).append(valueStr).toString();
					LOG.debug("inster sql : "+insertSql);
					String updateSql = new StringBuilder(UPDATE).append(table).append(setStr).append(where).toString();
					LOG.debug("update sql : "+updateSql);
					
					int n=0,m=0;
					insertPs = connection.prepareStatement(insertSql);
					updatePs = connection.prepareStatement(updateSql);
					ExecutorService pool = Executors.newFixedThreadPool(5);
					do{
						Object[] objects = null;
						objects = new Object[idColumns.length];
						for(int i=0;i<idColumns.length;i++){
							objects[i] = rs.getObject(idColumns[i]);
						}
						PreparedStatement countPs = this.find(connection, table, new QueryTermBean(afterWhere.toString(), objects), COUNT);
						ResultSet countRs = this.getResultSet(countPs);
						if(countRs.next()){
							if(countRs.getInt(1)==0){//新增
								for(int i=1;i<=columnCount;i++){
									String type = rsmd.getColumnTypeName(i).toLowerCase();
									if(Assert.isValidString(type)){
										if(type.indexOf(CHAR)>-1 || type.indexOf(TEXT)>-1){
											insertPs.setString(i, rs.getString(i));
										}else if(NUMBER.equals(type) || DECIMAL.equals(type)){
											insertPs.setBigDecimal(i, rs.getBigDecimal(i));
										}else if(type.indexOf(TIME)>-1 || type.indexOf(DATE)>-1){
											Timestamp timestamp = rs.getTimestamp(i);
											Timestamp nowTime = new Timestamp(System.currentTimeMillis());
											if(timestamp!=null && Assert.isValidString(modifyTimeColumn) 
													&& modifyTimeColumn.equalsIgnoreCase(rsmd.getColumnName(i))
													&& timestamp.after(nowTime)){//修正晚于当前时间的修改时间乱入
												insertPs.setTimestamp(i, nowTime);
											}else{
												insertPs.setTimestamp(i, timestamp);
											}
										}else if(type.indexOf(BLOB)>-1){
											insertPs.setBlob(i, rs.getBlob(i));
										}else if(CLOB.equals(type)){
											insertPs.setClob(i, rs.getClob(i));
										}else{
											insertPs.setObject(i, rs.getObject(i));
										}
									}
								}
								insertPs.addBatch();
								n++;
								if(n==EXECUTE_NUM){//先执行一批EXECUTE_NUM条
									pool.execute(new PSExecuter(insertPs,table,n));
									insertPs = connection.prepareStatement(insertSql);
									n=0;
								}
							}else{//更新
								int i=1;
								for(;i<=columnCount;i++){
									String type = rsmd.getColumnTypeName(i).toLowerCase();
									if(Assert.isValidString(type)){
										if(type.indexOf(CHAR)>-1 || type.indexOf(TEXT)>-1){
											updatePs.setString(i, rs.getString(i));
										}else if(NUMBER.equals(type) || DECIMAL.equals(type)){
											updatePs.setBigDecimal(i, rs.getBigDecimal(i));
										}else if(type.indexOf(TIME)>-1 || type.indexOf(DATE)>-1){
											Timestamp timestamp = rs.getTimestamp(i);
											Timestamp nowTime = new Timestamp(System.currentTimeMillis());
											if(timestamp!=null && Assert.isValidString(modifyTimeColumn) 
													&& modifyTimeColumn.equalsIgnoreCase(rsmd.getColumnName(i))
													&& timestamp.after(nowTime)){//修正晚于当前时间的修改时间乱入
												updatePs.setTimestamp(i, nowTime);
											}else{
												updatePs.setTimestamp(i, timestamp);
											}
										}else if(type.indexOf(BLOB)>-1){
											updatePs.setBlob(i, rs.getBlob(i));
										}else if(CLOB.equals(type)){
											updatePs.setClob(i, rs.getClob(i));
										}else{
											updatePs.setObject(i, rs.getObject(i));
										}
									}
								}
								if(Assert.isValidArray(idColumns)){
									for(String idColumn : idColumns){
										updatePs.setObject(i, rs.getObject(idColumn));
										i++;
									}
								}
								updatePs.addBatch();
								m++;
								if(m==EXECUTE_NUM){//先执行一批EXECUTE_NUM条
									pool.execute(new PSExecuter(updatePs,table,m));
									updatePs = connection.prepareStatement(updateSql);
									m=0;
								}
							}
						}
						c3p0Helper.closes(countPs, countRs);
					}while(rs.next());
					if(n>0){//最后可能有不足EXECUTE_NUM条的
						pool.execute(new PSExecuter(insertPs,table,n));
					}
					if(m>0){//最后可能有不足EXECUTE_NUM条的
						pool.execute(new PSExecuter(updatePs,table,m));
					}
					ThreadPoolShutdownUtil.shutdownPool(pool, 20, TimeUnit.HOURS);
				}
			}
			result = true;
		} catch (SQLException e) {
			e.printStackTrace();
		} finally{
			c3p0Helper.closes(insertPs);
			c3p0Helper.closes(updatePs);
		}
		
		return result;
	}
	
	/**
	 * 删除
	 * @param connection
	 * @param table
	 * @param afterWhere
	 * @return
	 */
	public int delete(Connection connection, String table, String afterWhere){
		int result = -1;
		if(connection==null || Assert.isInvalidString(table)){
			return result;
		}
		PreparedStatement ps=null;
		StringBuilder where = new StringBuilder();
		if(Assert.isValidString(afterWhere)){
			where.append(WHERE).append(afterWhere);
		}
		StringBuilder sql = new StringBuilder(DELETE).append(table).append(where);
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql.toString());
			result=ps.executeUpdate();//增删改
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			c3p0Helper.closes(ps);
		}
		return result;
	}
	
	/**
	 * 删除
	 * @param connection
	 * @param table
	 * @param queryBean 删除条件
	 * @return
	 */
	public int delete(Connection connection, String table, QueryTermBean queryBean){
		int result = -1;
		if(connection==null || Assert.isInvalidString(table)){
			return result;
		}
		PreparedStatement ps=null;
		StringBuilder where = new StringBuilder();
		if(queryBean!=null && Assert.isValidString(queryBean.getTermStr())){
			where.append(WHERE).append(queryBean.getTermStr());
		}
		StringBuilder sql = new StringBuilder(DELETE).append(table).append(where);
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql.toString());
			if(queryBean!=null && Assert.isValidArray(queryBean.getTermValue())){
				for(int i=0;i<queryBean.getTermValue().length;i++){
					ps.setObject((i+1), queryBean.getTermValue()[i]);
				}
			}
			result=ps.executeUpdate();//增删改
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			c3p0Helper.closes(ps);
		}
		return result;
	}
	
	/**
	 * 执行增删改
	 * @param connection
	 * @param sql
	 * @return
	 */
	public int execute(Connection connection, String sql){
		int result = -1;
		if(connection==null || Assert.isInvalidString(sql)){
			return result;
		}
		PreparedStatement ps=null;
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql);
			result=ps.executeUpdate();//增删改
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			c3p0Helper.closes(ps);
		}
		return result;
	}
	
	/**
	 * 执行增删改
	 * @param connection
	 * @param sql
	 * @param args
	 * @return
	 */
	public int execute(Connection connection, String sql, Object... args){
		int result = -1;
		if(connection==null || Assert.isInvalidString(sql)){
			return result;
		}
		PreparedStatement ps=null;
		LOG.debug(sql);
		try {
			ps = connection.prepareStatement(sql);
			if(Assert.isValidArray(args)){
				for(int i=0;i<args.length;i++){
					ps.setObject((i+1), args[i]);
				}
			}
			result=ps.executeUpdate();//增删改
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			c3p0Helper.closes(ps);
		}
		return result;
	}
	
	/**
	 * 清空表
	 * @param connection
	 * @param table
	 * @return
	 */
	public int truncate(Connection connection, String table){
		return this.execute(connection, TRUNCATE_TABLE+table);
	}
	
	/**
	 * 执行sql获取ResultSet
	 * @param preparedStatement
	 * @return
	 */
	public ResultSet getResultSet(PreparedStatement preparedStatement){
		ResultSet rs = null;
		if(preparedStatement != null){
			try {
				rs = preparedStatement.executeQuery();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return rs;
	}
	
	/**
	 * PreparedStatement执行者
	 * 
	 * @author HuangChen
	 *
	 */
	protected class PSExecuter implements Runnable{
		
		private PreparedStatement ps;
		private String table;
		private int num;
		public PSExecuter(PreparedStatement ps, String table, int num){
			this.ps = ps;
			this.table = table;
			this.num = num;
		}
		
		@Override
		public void run() {
			if(ps==null)return;
			try {
				ps.executeBatch();
			} catch (SQLException e) {
				e.printStackTrace();
			}finally{
				c3p0Helper.closes(ps);
			}
			LOG.info("-----table "+table+" sync "+num+" data completed");
		}
		
	}
	
}
